package com.aos.iot.socket.common.store.pubsub;

import com.aos.iot.socket.common.message.Message;
import io.netty.util.internal.PlatformDependent;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

public class RedissonPubSubStore<M extends Message> implements PubSubStore<M> {

    private final RedissonClient redissonPub;

    private final RedissonClient redissonSub;

    private final ConcurrentMap<String, Queue<Integer>> map = PlatformDependent.newConcurrentHashMap();

    public RedissonPubSubStore(RedissonClient redissonPub, RedissonClient redissonSub) {
        this.redissonPub = redissonPub;
        this.redissonSub = redissonSub;
    }

    @Override
    public void publish(String clusterName, M msg) {
        redissonPub.getTopic(clusterName).publish(msg);
    }

    @Override
    public void subscribe(String clusterName, PubSubListener<M> listener, Class<M> clazz) {
        RTopic topic = redissonSub.getTopic(clusterName);
        int regId = topic.addListener(clazz, (channel, msg) -> listener.onMessage(msg));

        Queue<Integer> list = map.get(clusterName);
        if (list == null) {
            list = new ConcurrentLinkedQueue<>();
            Queue<Integer> oldList = map.putIfAbsent(clusterName, list);
            if (oldList != null) {
                list = oldList;
            }
        }
        list.add(regId);
    }

    @Override
    public void unsubscribe(String clusterName) {
        Queue<Integer> regIds = map.remove(clusterName);
        RTopic topic = redissonSub.getTopic(clusterName);
        for (Integer id : regIds) {
            topic.removeListener(id);
        }
    }

    @Override
    public void shutdown() {

    }
}
